package com.tmt.im.service.websocket;

import com.tmt.im.common.pojo.MemberStatusPoJo;
import jakarta.websocket.Session;
import lombok.extern.slf4j.Slf4j;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.sql.Timestamp;
import java.time.Duration;
import java.time.OffsetDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;

/**
 * WebSocket Session管理器
 *
 * @Description TODO
 * @Author wuyi
 * @Date 2024/8/15 12:05
 * @Version 1.0
 **/
@Slf4j
@Component
public class SessionContextManager {

    private final ConcurrentHashMap<Long, SessionContext> userIdAndSessionMap = new ConcurrentHashMap<>();

    public static SessionContextManager self;

    /**
     * 心跳间隔
     */
    private final Duration heartBeatInterval;

    private Ignite ignite;

    public SessionContextManager(@Value("${websocket.heart-beat-interval}")
                                 Duration heartBeatInterval,
                                 @Qualifier("asyncTaskExecutor")
                                 Executor asyncTaskExecutor) {
        this.heartBeatInterval = heartBeatInterval;
        //启动心跳线程。
        asyncTaskExecutor.execute(this::heartBeatRun);
        self = this;
    }

    public void setIgnite(Ignite ignite) {
        this.ignite = ignite;
    }

    public void addSession(Long userId, Session session) {
        SessionContext sessionContext = new SessionContext(userId, session);
        SessionContext result = userIdAndSessionMap.putIfAbsent(userId, sessionContext);
        if (result != null && !Objects.equals(result.getSession(), sessionContext.getSession())) {
            //已经存在则替换之。
            userIdAndSessionMap.replace(userId, sessionContext);
            try {
                //并且关闭原来的连接。
                result.getSession().close();
            } catch (Exception ex) {
                log.debug("新连接踢掉旧连接(userId={})", userId);
            }
        }
    }

    public Session getSessionByUserId(Long userId) {
        SessionContext sessionContext = userIdAndSessionMap.get(userId);
        return sessionContext != null ? sessionContext.getSession() : null;
    }

    public void removeSession(Long userId) {
        userIdAndSessionMap.remove(userId);
    }

    public void updateHeartBeatTime(Long userId) {
        SessionContext sessionContext = userIdAndSessionMap.get(userId);
        sessionContext.setLatestHeartBeat(new Timestamp(System.currentTimeMillis()));
    }

    public void closeRealTimePush(Long userId) {
        IgniteCache<Long, MemberStatusPoJo> statusCache = ignite.cache(MemberStatusPoJo.CACHE_NAME);
        MemberStatusPoJo statusPoJo = statusCache.get(userId);
        if (statusPoJo != null) {
            statusPoJo.setSessionId(null);
            statusCache.replace(userId, statusPoJo);
        }
    }

    public void heartBeatRun() {
        while (!Thread.currentThread().isInterrupted()) {
            //等待10秒
            try {
                synchronized (userIdAndSessionMap) {
                    userIdAndSessionMap.wait(heartBeatInterval.toMillis());
                }
            } catch (InterruptedException ex) {
                log.debug(ex.getMessage(), ex);
            }
            //发送心跳信息
            sendPingMsg();
        }
    }

    private void sendPingMsg() {
        int parallelismThreshold = Math.min(userIdAndSessionMap.size(), 5);
        List<SessionContext> invalidSessionContexts = new ArrayList<>();
        userIdAndSessionMap.forEachValue(parallelismThreshold, context -> {
            Timestamp latestHeartBeat = context.getLatestHeartBeat();
            Timestamp now = new Timestamp(System.currentTimeMillis());
            if (now.getTime() - latestHeartBeat.getTime() >= heartBeatInterval.toMillis()) {
                int tryCount = 3;
                boolean sendOk = false;
                do {
                    try {
                        context.getSession().getAsyncRemote().sendText("ping");
                        context.setLatestHeartBeat(now);
                        sendOk = true;
                        break;
                    } catch (Exception ex) {
                        if (tryCount > 1) {
                            log.error("发送心跳帧失败，继续尝试...", ex);
                        }
                    }
                    try {
                        Thread.sleep(1000);
                    } catch (Exception ignored) {
                    }
                    tryCount--;
                } while (tryCount > 0);

                //发送失败, 则放入invalidSessionContexts列表中。
                if (!sendOk) {
                    invalidSessionContexts.add(context);
                }
            }
        });

        if (!CollectionUtils.isEmpty(invalidSessionContexts)) {
            //关闭发送失败的Session
            for (SessionContext invalidSessionContext : invalidSessionContexts) {
                try {
                    invalidSessionContext.getSession().close();
                } catch (Exception ex) {
                    log.debug("关闭心跳帧发送失败的连接出现异常", ex);
                }
            }
        }
    }
}
